package spendreport.service;

import java.io.Serializable;
import java.util.Date;
import java.util.Iterator;

import org.apache.flink.streaming.api.functions.source.FromIteratorFunction;

import spendreport.pojo.AccountTransation;

public class RandomNumberAccountTransactionSource extends FromIteratorFunction<AccountTransation> {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	public RandomNumberAccountTransactionSource() {
		super(new RandomNumberIterator());
		// TODO Auto-generated constructor stub
	}

	private static class RandomNumberIterator
			implements Iterator<AccountTransation>, Serializable{

		/**
		 * 
		 */
		private static final long serialVersionUID = 1L;
		private String[] accountIds = { "1001", "1002", "1003", "1004" };

		private RandomNumberIterator() {
		}

		@Override
		public boolean hasNext() {
			return true;
		}

		@Override
		public AccountTransation next() {
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
			int index = (int) (Math.random() * accountIds.length);
			String accountId = accountIds[index];
			Double money = Math.random() * 1000 + 1;
			AccountTransation accountTransation = new AccountTransation();
			accountTransation.setAccountId(accountId);
			accountTransation.setTransationTime(new Date());
			accountTransation.setTransationMoney(money);;
			return accountTransation;
		}
	}

}

